package stream;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class operator_partition {
    public static void main(String[] args) throws Exception {
    //   创建流式数据环
        StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();
    //    添加数据源
        DataStreamSource<Long> source = env.addSource(new oneparalleSource());

        //需要把数据类型转换，把 Long ----> Tuple1<Long>
        DataStream<Tuple1<Long>> data =  source.map(new MapFunction<Long, Tuple1<Long>>() {
            public Tuple1<Long> map(Long aLong) throws Exception {
                return new Tuple1<Long>(aLong);
            }
        });

        DataStream<Tuple1<Long>> result = data.partitionCustom(new Mypartitioner(), 0);

        result.map(new MapFunction<Tuple1<Long>, Long>() {
            public Long map(Tuple1<Long> longTuple1) throws Exception {
                Long data_ = longTuple1.f0;
                //System.out.println(longTuple1);
                //System.out.println("-----------");
                System.out.println("线程ID：" + Thread.currentThread().getId()+"\t 数据："+data_);
                return data_;


            }
        }).print();

        env.execute("operatro_filter");
    }

    static class Mypartitioner implements Partitioner<Long>{

        public int partition(Long aLong, int i) {
            if(aLong%2==0){
                return 0;
            }
            else{
                return 1;
            }
        }
    }
}

